package com.qf.listener;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class Consumer1 {

    /*
        接收到的消息类型必须和生产段的发送消息类型对应
           生产端是String，消费端String
           生产端是ProduceRecord，消费端ConsumerRecord
     */
    @KafkaListener(topics = "test")
    public void cons1(ConsumerRecord record /*, Acknowledgment ack*/){
        //服务器信息
        int partition = record.partition();
        long offset = record.offset();
        //数据(String)
        Object key = record.key();
        Object value = record.value();

        System.out.println("partition:" + partition + "==offset:" + offset);
        System.out.println("key:" + key + "==value:" + value);
        System.out.println("--------------------------------------------");

//        //确认提交
//        ack.acknowledge();
//        //不提交
//        ack.nack(0);
    }
}
